package net.bwie.realtime.warehouse.ADS;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.warehouse.ADS
 * @Author: zhangleqing
 * @CreateTime: 2025-09-05  10:36
 * @Description: TODO 退中检测
 * 根据每天分组 求 处理中退款金额 和 预计可挽回金额
 * 处理中退款金额=退款状态为申请中的退款金额
 * 预计可挽回金额 = 本次申请退款金额 ×（1 - 前 1 年本品类实际退款金额 / 前 1 年本品类总申请退款金额）
 *
 * 如果没有 “1 -”，直接用 “本次申请金额 × 历史退款率”，算出来的是 “预计会退给用户的钱”（而不是 “可挽回的钱”），完全偏离了目标。
 * @Version: 1.0
 */
public class Two {
    public static void main(String[] args) {
        // 1 开启上下文
        TableEnvironment tableEnv = getTableEnv();

        // 2 读取数据
        readTable(tableEnv);

        // 3 数据清洗+加工
        Table resultTable = handle(tableEnv);

//        resultTable.execute().print();

        // 4 映射表创建
        createView(tableEnv);

        // 5 输出数据
        saveToSink(tableEnv, resultTable);
    }

    private static void saveToSink(TableEnvironment tableEnv, Table resultTable) {
        tableEnv.createTemporaryView("resultTable", resultTable);
        tableEnv.executeSql(
                "insert into refund_two " +
                        " select * from resultTable"
        ).print();
    }

/*
写入Doris中
CREATE TABLE refund_two (
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3),

    processing_refund_amount DOUBLE,
    total_estimated_recoverable_amount DOUBLE,
    PRIMARY KEY (`window_start`) NOT ENFORCED
) WITH (
    'connector' = 'doris',
    'fenodes' = 'node102:8030',
    'table.identifier' = 'refund_insights.refund_two',
    'username' = 'root',
    'password' = '123456',
    // 1. 替代 sink.batch.size：达到1000条数据就触发写入（测试用，生产可改1000）
    'sink.buffer-flush.max-rows' = '1000',
    // 2. 替代 sink.batch.interval：10秒超时触发写入（防止数据积压）
    'sink.buffer-flush.interval' = '10000',
    // 3. 替代 sink.max-retries：失败重试3次（原参数支持，保留）
    'sink.max-retries' = '3',
    // 4. 适配 Upsert 语义：允许删除/更新（旧版通过该参数支持）
    'sink.enable-delete' = 'true',
    // 5. 数据格式：与Kafka输出匹配（原参数支持，保留）
    'sink.properties.format' = 'json',
    // 6. （可选）批量写入模式（提升性能，旧版支持）
    'sink.enable.batch-mode' = 'true'
)
 */

    private static void createView(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "CREATE TABLE refund_two (\n" +
                        "    window_start TIMESTAMP(3),\n" +
                        "    window_end TIMESTAMP(3),\n" +
                        "\n" +
                        "    processing_refund_amount DOUBLE,\n" +
                        "    total_estimated_recoverable_amount DOUBLE,\n" +
                        "    PRIMARY KEY (`window_start`) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'refund_insights.refund_two',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    // 1. 替代 sink.batch.size：达到1000条数据就触发写入（测试用，生产可改1000）\n" +
                        "    'sink.buffer-flush.max-rows' = '1000',\n" +
                        "    // 2. 替代 sink.batch.interval：10秒超时触发写入（防止数据积压）\n" +
                        "    'sink.buffer-flush.interval' = '10000',\n" +
                        "    // 3. 替代 sink.max-retries：失败重试3次（原参数支持，保留）\n" +
                        "    'sink.max-retries' = '3',\n" +
                        "    // 4. 适配 Upsert 语义：允许删除/更新（旧版通过该参数支持）\n" +
                        "    'sink.enable-delete' = 'true',\n" +
                        "    // 5. 数据格式：与Kafka输出匹配（原参数支持，保留）\n" +
                        "    'sink.properties.format' = 'json',\n" +
                        "    // 6. （可选）批量写入模式（提升性能，旧版支持）\n" +
                        "    'sink.enable.batch-mode' = 'true'\n" +
                        ")"
        );
    }

/*
select
    -- 统计日期
    TUMBLE_START(apply_time, INTERVAL '1' DAY) AS window_start,
    TUMBLE_END(apply_time, INTERVAL '1' DAY) AS window_end,
    product_category,

    -- 处理中退款金额：当日发起且当前未完结的退款金额总和
    SUM(CASE WHEN refund_status = '处理中' THEN refund_amount ELSE 0 END)
    AS processing_refund_amount

    -- 预计可挽回金额=退款金额×(前1年本品类退款金额退款金额−实际退款金额×100%)

from dwd_refund_wide
GROUP BY
    TUMBLE(apply_time, INTERVAL '1' DAY),
    product_category


UNIX_TIMESTAMP 转换成时间戳
字段类型需要转换为字符串

-- 基于已有宽表计算所需指标
WITH history_stats AS (
    SELECT
        *,
        -- 前1年本品类总申请退款金额
        SUM(refund_amount) OVER (
            PARTITION BY product_category
            ORDER BY apply_time
            RANGE BETWEEN INTERVAL '1' YEAR PRECEDING AND CURRENT ROW
        ) AS total_refund_last_year,
        -- 前1年本品类实际退款金额（仅申请成功）
        SUM(CASE WHEN refund_status = '申请成功' THEN refund_amount ELSE 0 END) OVER (
            PARTITION BY product_category
            ORDER BY apply_time
            RANGE BETWEEN INTERVAL '1' YEAR PRECEDING AND CURRENT ROW
        ) AS actual_refund_last_year
    FROM dwd_refund_wide
),
refund_with_calculation AS (
    SELECT
        *,
        -- 基于历史统计计算可挽回金额
        refund_amount * (
            1 - CASE WHEN total_refund_last_year = 0
                THEN 0
                ELSE actual_refund_last_year / total_refund_last_year
            END
        ) AS estimated_recoverable
    FROM history_stats
)
-- 按天统计结果
SELECT
    TUMBLE_START(apply_time, INTERVAL '1' DAY) AS window_start,
    TUMBLE_END(apply_time, INTERVAL '1' DAY) AS window_end,
    product_category,
    SUM(CASE WHEN refund_status = '申请中' THEN refund_amount ELSE 0 END) AS processing_refund_amount,
    ROUND(SUM(estimated_recoverable), 4) AS total_estimated_recoverable_amount
FROM refund_with_calculation
GROUP BY
    TUMBLE(apply_time, INTERVAL '1' DAY),
    product_category


优化版本
WITH history_stats AS (
    SELECT
        product_category,  -- 窗口分组依赖
        apply_time,        -- 排序和窗口聚合依赖
        refund_amount,     -- 求和计算依赖
        refund_status,     -- 条件判断依赖
        -- 前1年本品类总申请退款金额
        SUM(refund_amount) OVER (
            PARTITION BY product_category
            ORDER BY apply_time
            RANGE BETWEEN INTERVAL '1' YEAR PRECEDING AND CURRENT ROW
        ) AS total_refund_last_year,
        -- 前1年本品类实际退款金额（仅申请成功）
        SUM(CASE WHEN refund_status = '申请成功' THEN refund_amount ELSE 0 END) OVER (
            PARTITION BY product_category
            ORDER BY apply_time
            RANGE BETWEEN INTERVAL '1' YEAR PRECEDING AND CURRENT ROW
        ) AS actual_refund_last_year
    FROM dwd_refund_wide
),
refund_with_calculation AS (
    SELECT
        apply_time,  -- 后续窗口聚合依赖
        refund_status,  -- 条件判断依赖
        refund_amount,  -- 条件判断依赖
        total_refund_last_year,
        actual_refund_last_year,
        -- 基于历史统计计算可挽回金额
        refund_amount * (
            1 - CASE WHEN total_refund_last_year = 0
                THEN 0
                ELSE actual_refund_last_year / total_refund_last_year
            END
        ) AS estimated_recoverable
    FROM history_stats
)
-- 按天统计结果
SELECT
    TUMBLE_START(apply_time, INTERVAL '1' DAY) AS window_start,
    TUMBLE_END(apply_time, INTERVAL '1' DAY) AS window_end,
    SUM(CASE WHEN refund_status = '申请中' THEN refund_amount ELSE 0 END) AS processing_refund_amount,
    ROUND(SUM(estimated_recoverable), 4) AS total_estimated_recoverable_amount
FROM refund_with_calculation
GROUP BY
    TUMBLE(apply_time, INTERVAL '1' DAY)

 */

    private static Table handle(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(
                "WITH history_stats AS (\n" +
                        "    SELECT\n" +
                        "        *, \n" +
                        "        -- 前1年本品类总申请退款金额\n" +
                        "        SUM(refund_amount) OVER (\n" +
                        "            PARTITION BY product_category\n" +
                        "            ORDER BY apply_time\n" +
                        "            RANGE BETWEEN INTERVAL '1' YEAR PRECEDING AND CURRENT ROW\n" +
                        "        ) AS total_refund_last_year, \n" +
                        "        -- 前1年本品类实际退款金额（仅申请成功）\n" +
                        "        SUM(CASE WHEN refund_status = '申请成功' THEN refund_amount ELSE 0 END) OVER (\n" +
                        "            PARTITION BY product_category\n" +
                        "            ORDER BY apply_time\n" +
                        "            RANGE BETWEEN INTERVAL '1' YEAR PRECEDING AND CURRENT ROW\n" +
                        "        ) AS actual_refund_last_year\n" +
                        "    FROM dwd_refund_wide\n" +
                        // 定义时间窗口为 “当前行的申请时间之前 1 年” 到 “当前行”（包含当前行）。
                        "),\n" +
                        "refund_with_calculation AS (\n" +
                        "    SELECT\n" +
                        "        *, \n" +
                        "        -- 基于历史统计计算可挽回金额\n" +
                        "        refund_amount * (\n" +
                        "            1 - CASE WHEN total_refund_last_year = 0 \n" +
                        "                THEN 0 \n" +
                        "                ELSE actual_refund_last_year / total_refund_last_year \n" +
                        "            END\n" +
                        "        ) AS estimated_recoverable\n" +
                        "    FROM history_stats\n" +
                        ")\n" +
                        // 基于第一层计算出的历史数据，计算每笔退款的预计可挽回金额
                        // 使用公式：退款金额 × (1 - 历史退款率) 来计算可挽回金额
                        "-- 按天统计结果\n" +
                        "SELECT\n" +
                        "    TUMBLE_START(apply_time, INTERVAL '1' DAY) AS window_start,\n" +
                        "    TUMBLE_END(apply_time, INTERVAL '1' DAY) AS window_end,\n" +
                        "    SUM(CASE WHEN refund_status = '申请中' THEN refund_amount ELSE 0 END) AS processing_refund_amount,\n" +
                        "    ROUND(SUM(estimated_recoverable), 4) AS total_estimated_recoverable_amount\n" +
                        "FROM refund_with_calculation\n" +
                        "GROUP BY\n" +
                        "    TUMBLE(apply_time, INTERVAL '1' DAY)"
                        // 按天进行分组统计（使用滚动窗口）
                        // 计算每天的处理中退款金额总和
                        // 计算每天的预计可挽回金额总和
        );
        // 先求 前1年本品类总申请退款金额 和 前1年本品类实际退款金额（仅申请成功） 然后按照公式计算可挽回金额，最后开一天窗口求【申请中退款金额和可挽回金额】
        return table;
    }

/*
一个问题 水位线设置的时间不能为空，如果后面想要使用水位线，那就前面加上水位线字段不为空，确保
后面水位线正确
 */

    private static void readTable(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "create table dwd_refund_wide (\n" +
                        "    -- 退款明细信息\n" +
                        "    refund_detail_id STRING,\n" +
                        "    refund_id STRING,\n" +
                        "    order_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    apply_time TIMESTAMP(3),\n" +
                        "    refund_status STRING,\n" +
                        "    finish_time TIMESTAMP(3),\n" +
                        "    product_id STRING,\n" +
                        "    refund_amount DOUBLE,\n" +
                        "    user_fill_reason STRING,\n" +
                        "    refund_scene STRING,\n" +
                        "\n" +
                        "    -- 订单基础信息\n" +
                        "    shop_id STRING,\n" +
                        "    pay_time TIMESTAMP(3),\n" +
                        "    promise_ship_time TIMESTAMP(3),\n" +
                        "    click_ship_time TIMESTAMP(3),\n" +
                        "    order_amount DECIMAL(10,2),\n" +
                        "    order_status STRING,\n" +
                        "\n" +
                        "    -- 商品信息\n" +
                        "    product_category STRING,\n" +
                        "    product_sku STRING,\n" +
                        "    product_price DOUBLE,\n" +
                        "    product_desc STRING,\n" +
                        "    proctime AS PROCTIME(),\n" +
                        "    WATERMARK FOR apply_time AS apply_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'dwd_refund_wide',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'dwd_refund_wide',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
    }

    public static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态TTL设置为25小时（3600*25=90000秒），覆盖24小时的关联区间
        configuration.setString("table.exec.state.ttl", "90000 s");
//        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }
}
